package com.kiss.rpc.core;

import com.kiss.rpc.NettyBootstrapInitializer;
import com.kiss.rpc.RpcBootstrap;
import com.kiss.rpc.compress.CompressorFactory;
import com.kiss.rpc.discovery.Registry;
import com.kiss.rpc.enumeration.RequestType;
import com.kiss.rpc.serialize.SerializerFactory;
import com.kiss.rpc.transport.message.RpcRequest;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


@Slf4j
public class HeartbeatDetector {
    public static AtomicBoolean isDetecting = new AtomicBoolean(true);
    public static void detectHeartbeat(String serviceName) {
        Registry registry = RpcBootstrap.getInstance().getConfig().getRegistryConfig().getRegistry();
        List<InetSocketAddress> addressList = registry.lookup(serviceName,
                RpcBootstrap.getInstance().getConfig().getGroup());

        for (InetSocketAddress address : addressList) {
            if (!RpcBootstrap.CHANNEL_CACHE.containsKey(address)) {
                try {
                    Channel channel = NettyBootstrapInitializer.getBootstrap().connect(address).sync().channel();
                    RpcBootstrap.CHANNEL_CACHE.put(address, channel);
                } catch (Exception e) {
                    log.error("{}接口从注册中心共发现{}个可用地址，为{}地址创建channel失败，服务已不可用，将从注册中心中删除此节点",
                            serviceName, addressList.size(), address);
                    registry.unregister(serviceName, RpcBootstrap.getInstance().getConfig().getGroup(), address.toString());
                }
            }
        }

        if (isDetecting.get()) {
            if (log.isDebugEnabled()) {
                log.debug("已经有心跳检测线程对所有的缓存Channel进行心跳检测， 不创建新的");
            }
            return;
        }

        isDetecting.set(true);
        Thread thread = new Thread(() ->
                new Timer().scheduleAtFixedRate(new MyTimerTask(), 0, 10000),
                "rpc-HeartbeatDetector-Thread");
        thread.setDaemon(true);
        thread.start();
    }

    private static class MyTimerTask extends TimerTask {

        @Override
        public void run() {
            RpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.clear();

            Map<InetSocketAddress, Channel> cache = RpcBootstrap.CHANNEL_CACHE;

            for (Map.Entry<InetSocketAddress, Channel> entry : cache.entrySet()) {
                int tryTimes = 3;
                while (tryTimes > 0) {
                    Channel channel = entry.getValue();

                    long start = System.currentTimeMillis();

                    RpcRequest rpcRequest = RpcRequest.builder()
                            .requestId(RpcBootstrap.getInstance().getConfig().getIdGenerator().getId())
                            .compressType(CompressorFactory.getCompressor(RpcBootstrap.getInstance().getConfig()
                                    .getCompressType()).getCode())
                            .serializeType(SerializerFactory.getSerializer(RpcBootstrap.getInstance().getConfig()
                                    .getSerializeType()).getCode())
                            .requestType(RequestType.HEART_BEAT.getId())
                            .timeStamp(start)
                            .build();

                    CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                    channel.writeAndFlush(rpcRequest).addListener(future -> {
                        if (!future.isSuccess()) {
                            completableFuture.completeExceptionally(future.cause());
                        } else {
                            completableFuture.complete(future.get());
                        }
                    });

                    long endTime;
                    try {
                        completableFuture.get(1, TimeUnit.SECONDS);
                        endTime = System.currentTimeMillis();
                    } catch (Exception e) {
                        tryTimes--;
                        log.error("和地址为【{}】的主机连接发生异常.正在进行第【{}】次重试......",
                                channel.remoteAddress(), 3 - tryTimes);

                        if (tryTimes == 0) {
                            RpcBootstrap.CHANNEL_CACHE.remove(entry.getKey());
                        }

                        try {
                            Thread.sleep(10 * (new Random().nextInt(5)));
                        } catch (InterruptedException e1) {
                            throw new RuntimeException(e1);
                        }

                        continue;
                    }

                    long time = endTime - start;
                    // 使用treemap进行缓存
                    RpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.put(time, channel);
                    log.debug("和[{}]服务器的响应时间是[{}].", entry.getKey(), time);
                    break;
                }
            }

            log.info("-----------------------响应时间的treemap----------------------");
            for (Map.Entry<Long, Channel> entry : RpcBootstrap.ANSWER_TIME_CHANNEL_CACHE.entrySet()) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}]--->channelId:[{}]", entry.getKey(), entry.getValue().id());
                }
            }
        }
    }
}
